package com.yeming.flink.practice.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer

object KafkaSourceString {

  def main(args: Array[String]): Unit = {
    //初始化环境变量
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //修改并行度
    streamEnv.setParallelism(1)
    //设置数据源,读取HDFS文件
    val props = new Properties()
    props.setProperty("bootstrap.servers", "f2:9092,f3:9092,f4:9092")
    props.setProperty("group.id", "fink01")
    props.setProperty("key.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("value.deserializer", classOf[StringDeserializer].getName)
    props.setProperty("auto.offset.reset", "latest")

    val stream: DataStream[String] = streamEnv.addSource(new FlinkKafkaConsumer[String]("flinktest01", new SimpleStringSchema(), props))
    //Transformation
    //Sink
    stream.print()
    streamEnv.execute("")
  }
}
